package com.bwstudio.demo.provider.message;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.bwstudio.demo.provider.dao.AccountInfoDao;
import com.bwstudio.demo.provider.model.AccountChangeEvent;
import com.bwstudio.demo.provider.service.AccountInfoService;
import com.bwstudio.demo.provider.service.NewAccountInfoService;
import com.google.gson.Gson;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * @Description: 业务方法中执行 sendMessageInTransaction 时对应 txProducerGroup 的事务消息的监听器
 * txProducerGroup = "producer_group_txmsg_bank1" 这个group的作用范围在本应用的进程内，作为应用内部发送事务消息的监听目标
 * @Author: Bu Wei
 * @Date: 2020年3月11日 上午10:49:37
 */
@Component
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
//@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_bank1")
@RocketMQTransactionListener
public class NewProducerTxmsgListener implements RocketMQLocalTransactionListener {

    private final NewAccountInfoService newAccountInfoService;

    private final AccountInfoDao accountInfoDao;

    /**
     * @param message
     * @param o
     * @return
     * @Description: 事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调
     * @Author: Bu Wei
     * @Date: 2020年3月11日 上午10:56:07
     */
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        log.info("执行本地事务");
        MessageHeaders headers = message.getHeaders();
        //获取事务ID
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        Integer accountInfoId = Integer.valueOf((String) headers.get("accountInfo_id"));
        log.info("transactionId is {}, accountInfo_id is {}", transactionId, accountInfoId);

        try {
            // 解析message，转成AccountChangeEvent
            String messageString = new String((byte[]) message.getPayload());
            Gson gson = new Gson();
            AccountChangeEvent accountChangeEvent = gson.fromJson(messageString, AccountChangeEvent.class);

            // 执行本地事务，扣减金额
            newAccountInfoService.doReduceAccountBalance(accountChangeEvent);

            // 当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * @param message
     * @return
     * @Description: 当一阶段half message发送成功，丢失了二阶段确认消息时，
     * 进行半消息事务状态回查，查询本地事务是否执行成功（是否扣减金额）
     * @Author: Bu Wei
     * @Date: 2020年3月11日 上午10:57:54
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        MessageHeaders headers = message.getHeaders();
        //获取事务ID
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        log.info("检查本地事务,事务ID:{}", transactionId);
        //根据事务id从日志表检索
        int existTx = accountInfoDao.isExistTx(transactionId);
        if (existTx > 0) {
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}